package com.hisense.ovcloud.mqtt.warehouse.service;

import cn.hutool.core.util.ObjectUtil;
import com.google.common.collect.Lists;
import com.hisense.ovcloud.commons.dto.common.DomainInfoDto;
import com.hisense.ovcloud.commons.dto.common.ProductInfoInCache;
import com.hisense.ovcloud.commons.dto.common.mqtt.MqttDeviceInfoInCache;
import com.hisense.ovcloud.commons.dto.common.mqtt.MqttProductInfoInCache;
import com.hisense.ovcloud.commons.dto.req.*;
import com.hisense.ovcloud.commons.dto.resp.*;
import com.hisense.ovcloud.commons.tools.DkeyTool;
import com.hisense.ovcloud.commons.tools.LocalLFUCacheUtil;
import com.hisense.ovcloud.commons.tools.ZCServerRequestUtil;
import com.hisense.ovcloud.mqtt.warehouse.Constants;
import com.hisense.ovcloud.mqtt.warehouse.config.WarehouseExecutor;
import com.hisense.ovcloud.mqtt.warehouse.config.WarehouseProperties;
import com.hisense.ovcloud.mqtt.warehouse.dao.ProductDao;
import com.hisense.ovcloud.mqtt.warehouse.dao.WarehouseDao;
import com.hisense.ovcloud.mqtt.warehouse.entity.device.ProductInfo;
import com.hisense.ovcloud.mqtt.warehouse.entity.warehouse.MqttDeviceWarehouse;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Service;
import org.springframework.web.reactive.function.server.ServerRequest;
import reactor.core.publisher.Mono;

import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;

@Service
@Slf4j
public class DevicesService {
    private final WarehouseProperties warehouseProperties;
    private final WarehouseExecutor warehouseExecutor;

    private final ProductDao productDao;

    private final LocalLFUCacheUtil<MqttProductInfoInCache> mqttProductInfoInCaches;

    private final LocalLFUCacheUtil<MqttDeviceInfoInCache> mqttDeviceInfoInCaches;

    private final LocalLFUCacheUtil<ProductInfoInCache> productInfoInCaches;


    public final WarehouseDao warehouseDao;

    private final LocalLFUCacheUtil<DomainInfoDto> fetureCodeCaches;


    public DevicesService(WarehouseProperties warehouseProperties, WarehouseExecutor warehouseExecutor, @Qualifier("mqttProductInfoInCache") LocalLFUCacheUtil<MqttProductInfoInCache> mqttProductInfoInCaches, WarehouseDao warehouseDao, @Qualifier("mqttDeviceInfoInCache") LocalLFUCacheUtil<MqttDeviceInfoInCache> mqttDeviceInfoInCaches, @Qualifier("productInfoInCache") LocalLFUCacheUtil<ProductInfoInCache> productInfoInCaches, ProductDao productDao, @Qualifier("featureCodeInCache") LocalLFUCacheUtil<DomainInfoDto> featureCodeInCache) {
        this.warehouseProperties = warehouseProperties;
        this.warehouseExecutor = warehouseExecutor;
        this.mqttProductInfoInCaches = mqttProductInfoInCaches;
        this.warehouseDao = warehouseDao;
        this.mqttDeviceInfoInCaches = mqttDeviceInfoInCaches;
        this.productInfoInCaches = productInfoInCaches;
        this.productDao = productDao;
        this.fetureCodeCaches = featureCodeInCache;
    }

    private boolean checkIfMQTTDevices(ProductInfo productInfo){
        return StringUtils.isNotEmpty(productInfo.getCommunication()) && !StringUtils.containsAnyIgnoreCase(productInfo.getCommunication(), "julink");
    }

    public Mono<Boolean> createSqlTable(ServerRequest req) {
        return req.bodyToMono(CreateSqlReq.class)
                .flatMap(r -> {
                    List<String> sqlList = new ArrayList<>(List.of(
                            Constants.Warehouse.CREATE_DEVICE_STATUS_TABLE.replaceAll(Constants.Warehouse.DOMAIN_ID_TO_REPLACE, r.getDomainId()),
                            Constants.Warehouse.CREATE_DEVICE_WAREHOUSE_TABLE.replaceAll(Constants.Warehouse.DOMAIN_ID_TO_REPLACE, r.getDomainId()),
                            Constants.Warehouse.CREATE_DEVICE_PROPERTY_TABLE.replaceAll(Constants.Warehouse.DOMAIN_ID_TO_REPLACE, r.getDomainId()),
                            Constants.Warehouse.CREATE_DEVICE_LICENSE_TABLE.replaceAll(Constants.Warehouse.DOMAIN_ID_TO_REPLACE, r.getDomainId()),
                            Constants.Warehouse.CREATE_DEVICE_FAILED_LICENSE_TABLE.replaceAll(Constants.Warehouse.DOMAIN_ID_TO_REPLACE, r.getDomainId()),
                            Constants.Warehouse.CREATE_DEVICE_ATOM_TABLE.replaceAll(Constants.Warehouse.DOMAIN_ID_TO_REPLACE, r.getDomainId())
                    ));

                    return Mono.just(sqlList);
                }).zipWhen(sqls -> Mono.just(warehouseExecutor.directExecuteSQLBatch(sqls)), (x, y) -> y);
    }

    public Mono<RegisterDevicesResp> registerDevices(ServerRequest req) {
        String domain = ZCServerRequestUtil.getHeader(req, Constants.ZC.HEAHERS.HEADER_ZC_DOMAIN);
        String subDomain = ZCServerRequestUtil.getHeader(req, Constants.ZC.HEAHERS.HEADER_ZC_SUBDOMAIN);
        ZCServerRequestUtil.getHeader(req, Constants.ZC.HEAHERS.HEADER_ZC_DEVELOPER_ID);
        return req.bodyToMono(MqttRegisterDevicesReq.class)
                .flatMap(r -> fetchProductInfoInCaches(domain, subDomain).zipWhen(productInfoCache -> Mono.just(r.getDevices().stream().map(d -> {
                                    if(productInfoCache.getOneMachineOneSecret() != Constants.ProductInfos.OneMachineOneSecret.YES){
                                        AtomicReference<MqttDeviceWarehouse> ref = new AtomicReference<>(MqttDeviceWarehouse.builder().build());
                                        fetchMqttProductLevelInfoFromCache(r.getDomain(), r.getSubDomain())
                                                .subscribe(res -> ref.set(MqttDeviceWarehouse.builder()
                                                        .device_id(d.getDeviceId())
                                                        .sub_domain(subDomain)
                                                        .status(Constants.Warehouse.NOT_ACTIVATED)
                                                        .device_type(Constants.ProductInfos.DEFAULT_DEVICE_TYPE)
                                                        .batch_code(d.getBatchCode())
                                                        .public_key(Optional.ofNullable(res).orElse(MqttProductInfoInCache.builder().build()).getPublicKey())
                                                        .build()));
                                        return ref.get();

                                    }else{
                                        return MqttDeviceWarehouse.builder()
                                                .device_id(d.getDeviceId())
                                                .sub_domain(subDomain)
                                                .status(Constants.Warehouse.NOT_ACTIVATED)
                                                .device_type(Constants.ProductInfos.DEFAULT_DEVICE_TYPE)
                                                .batch_code(d.getBatchCode())
                                                .public_key(d.getPublicKey())
                                                .build();
                                    }
                                }).collect(Collectors.toList())), (x, y) -> y)
                                .map(totalList -> Pair.of(totalList, Lists.partition(totalList, warehouseProperties.getImportBatchSize())
                                        .parallelStream().map(subList -> zipSql(r.getDomain(), subList)).toList()))
                                .flatMap(pair -> {
                                    warehouseExecutor.directExecuteSQLBatchV2(pair.getValue());
                                    return Mono.just(pair.getKey());
                                })
                                .zipWhen(totalList -> {
                                    mqttDeviceInfoInCaches.batchDelete(totalList.stream().map(MqttDeviceWarehouse::getDevice_id).collect(Collectors.toList()));
                                    return Mono.just(RegisterDevicesResp.builder().succCount(totalList.size()).build());
                                },(x, y) -> y)
                );
    }

    public Mono<Boolean> registerProduct(ServerRequest req){
        String domain = ZCServerRequestUtil.getHeader(req, Constants.ZC.HEAHERS.HEADER_ZC_DOMAIN);
        String subDomain = ZCServerRequestUtil.getHeader(req, Constants.ZC.HEAHERS.HEADER_ZC_SUBDOMAIN);
        return req.bodyToMono(RegisterProductReq.class)
                .flatMap(r -> Mono.just(warehouseDao.registerProduct(domain, subDomain, r.getPublicKey(), r.getPrivateKey())));
    }

    public Mono<GenerateAesKeyResp> generateAesKey(ServerRequest req){
        ZCServerRequestUtil.getHeader(req, Constants.ZC.HEAHERS.HEADER_ZC_DOMAIN);
        ZCServerRequestUtil.getHeader(req, Constants.ZC.HEAHERS.HEADER_ZC_SUBDOMAIN);
        ZCServerRequestUtil.getHeader(req, Constants.ZC.HEAHERS.HEADER_ZC_DEVELOPER_ID);
        return Mono.just(GenerateAesKeyResp.builder().key(DkeyTool.generateRandomKey(Constants.ZC.ZC_HS_SESSION_KEY_LEN)).build());
    }

    public Mono<GenerateAesKeyAndFetchPublicKeyResp> generateAesKeyAndFetchPublicKey(ServerRequest req){
        return getPublicKey(req).zipWhen(getPublicKeyResp -> Mono.just(GenerateAesKeyAndFetchPublicKeyResp.builder()
                .key(DkeyTool.generateRandomKey(Constants.ZC.ZC_HS_SESSION_KEY_LEN))
                .pubkey(getPublicKeyResp.getPublicKey())
                .build()), (x, y) -> y);
    }

    public Mono<AuthBatchResp> authBatch(ServerRequest req){
        return req.bodyToMono(AuthBatchReq.class)
                .flatMap(r -> Mono.just(fetchDomainInfoByWifiId(r.getWifiId()))
                                .zipWhen(domainInfoDto -> fetchDeviceInfoFromCache(domainInfoDto.getDomain(), domainInfoDto.getSubDomain(), r.getWifiId())
                                        .zipWhen(mqttDeviceInfoInCache -> {
                                            if(StringUtils.isNotEmpty(mqttDeviceInfoInCache.getPublicKey())){
                                                return Mono.just(AuthBatchResp.builder().flag(true).build());
                                            }
                                            return Mono.just(AuthBatchResp.builder().build());
                                        }, (x, y) -> y), (x, y) -> y));
    }


    public Mono<GenerateAesKeyAndFetchPublicKeyResp> getDeviceInfoByWifiId(ServerRequest req){
        return getPublicKey(req).zipWhen(getPublicKeyResp -> Mono.just(GenerateAesKeyAndFetchPublicKeyResp.builder()
                .key(DkeyTool.generateRandomKey(Constants.ZC.ZC_HS_SESSION_KEY_LEN))
                .pubkey(getPublicKeyResp.getPublicKey())
                .build()), (x, y) -> y);
    }

    public Mono<GetPublicKeyResp> getPublicKey(ServerRequest req){
        String domain = ZCServerRequestUtil.getHeader(req, Constants.ZC.HEAHERS.HEADER_ZC_DOMAIN);
        String subDomain = ZCServerRequestUtil.getHeader(req, Constants.ZC.HEAHERS.HEADER_ZC_SUBDOMAIN);
        return req.bodyToMono(GetPublicKeyReq.class)
                .flatMap(r ->fetchDeviceInfoFromCache(domain, subDomain, r.getPhysicalDeviceId())
                        .zipWhen(mqttDeviceInfoInCache -> {
                            if(StringUtils.isNotEmpty(mqttDeviceInfoInCache.getPublicKey())){
                                return Mono.just(GetPublicKeyResp.builder().publicKey(mqttDeviceInfoInCache.getPublicKey()).build());
                            }
                            return Mono.just(GetPublicKeyResp.builder().build());
                        }, (x, y) -> y));
    }

    public Mono<GetPublicKeyResp> getPublicKeyInner(Mono<GetPublicKeyReq> req){
        return req.flatMap(r ->fetchDeviceInfoFromCache(r.getDomain(), r.getSubDomain(), r.getPhysicalDeviceId())
                        .zipWhen(mqttDeviceInfoInCache -> {
                            if(StringUtils.isNotEmpty(mqttDeviceInfoInCache.getPublicKey())){
                                return Mono.just(GetPublicKeyResp.builder().publicKey(mqttDeviceInfoInCache.getPublicKey()).build());
                            }
                            return Mono.just(GetPublicKeyResp.builder().build());
                        }, (x, y) -> y));
    }

    public DomainInfoDto fetchDomainInfoByWifiId(String wifiId){
        String key = wifiId.substring(0, 24);
        DomainInfoDto dto = fetureCodeCaches.getFromCache(key);
        if(ObjectUtil.isNotNull(dto)){
            return dto;
        }else{
            ProductInfo productInfo = productDao.findProductInfoByFeatureCode(key);
            if (null != productInfo && StringUtils.isNotEmpty(productInfo.getDomain())) {
                dto = DomainInfoDto.builder().domain(productInfo.getDomain()).subDomain(productInfo.getSubDomainId()).build();
                fetureCodeCaches.saveToCache(key, dto);
                return dto;
            }
            return null;
        }
    }

    private Mono<MqttProductInfoInCache> fetchMqttProductLevelInfoFromCache(String domain, String subDomain) {
        return Mono.just(Optional.ofNullable(mqttProductInfoInCaches.getFromCache(String.format("%s-%s", domain, subDomain))).orElse(MqttProductInfoInCache.builder().build())).zipWhen(inCache -> {
            if (StringUtils.isNotEmpty(inCache.getDomain()) || StringUtils.isNotEmpty(inCache.getSubDomain())) {
                return Mono.just(inCache);
            } else {
                return Mono.just(warehouseDao.getWarehouseProductStatus(domain, subDomain)).zipWhen(warehouseProductStatus -> {
                    if (warehouseProductStatus != null) {
                        return Mono.just(MqttProductInfoInCache.builder()
                                .domain(warehouseProductStatus.getDomain())
                                .subDomain(warehouseProductStatus.getSubDomain())
                                .publicKey(warehouseProductStatus.getPublicKey())
                                .privateKey(warehouseProductStatus.getPrivateKey())
                                .regType(warehouseProductStatus.getType())
                                .proStatus(warehouseProductStatus.getStatus())
                                .build()).zipWhen(newInfo -> {
                            mqttProductInfoInCaches.saveToCache(String.format("%s-%s", domain, subDomain), newInfo);
                            return Mono.just(newInfo);
                        }, (x, y) -> y);
                    }
                    return Mono.just(MqttProductInfoInCache.builder().build());
                }, (x, y) -> y);
            }
        }, (x, y) -> y);
    }

    public Mono<Boolean> updateDeviceActiveStatus(ServerRequest req) {
        String domain = ZCServerRequestUtil.getHeader(req, Constants.ZC.HEAHERS.HEADER_ZC_DOMAIN);
        return req.bodyToMono(UpdateDeviceStatusReq.class)
                .flatMap(r ->Mono.just(Optional.ofNullable(mqttDeviceInfoInCaches.getFromCache(r.getDeviceId()))
                        .orElse(MqttDeviceInfoInCache.builder().build())).zipWhen(inCache -> {
                    if (StringUtils.isNotEmpty(inCache.getWifiId()) && inCache.getStatus() == Constants.Warehouse.NOT_ACTIVATED) {
                        return Mono.just(warehouseDao.updateDeviceActivateStatus(domain, r.getDeviceId()))
                                .zipWhen(res -> Mono.just(updateStatusInCache(r.getDeviceId())), (x, y) -> x);
                    } else {
                        return Mono.just(true);
                    }
                }, (x, y) -> y));
    }

    public Mono<Boolean> updateDeviceActiveStatusInner(Mono<UpdateDeviceStatusReq> req) {
        return req.flatMap(r ->Mono.just(Optional.ofNullable(mqttDeviceInfoInCaches.getFromCache(r.getDeviceId()))
                        .orElse(MqttDeviceInfoInCache.builder().build())).zipWhen(inCache -> {
                    if (StringUtils.isNotEmpty(inCache.getWifiId()) && inCache.getStatus() == Constants.Warehouse.NOT_ACTIVATED) {
                        return Mono.just(warehouseDao.updateDeviceActivateStatus(String.valueOf(r.getDomain()), r.getDeviceId()))
                                .zipWhen(res -> Mono.just(updateStatusInCache(r.getDeviceId())), (x, y) -> x);
                    } else {
                        return Mono.just(true);
                    }
                }, (x, y) -> y));
    }


    private Mono<MqttDeviceInfoInCache> fetchDeviceInfoFromCache(String domain, String subDomain, String wifiId) {
        return Mono.just(Optional.ofNullable(mqttDeviceInfoInCaches.getFromCache(wifiId))
                .orElse(MqttDeviceInfoInCache.builder().build())).zipWhen(inCache -> {
            if (StringUtils.isNotEmpty(inCache.getDomain()) ) {
                return Mono.just(inCache);
            } else {
                return Mono.just(warehouseDao.getByWifiId(domain, wifiId)).zipWhen(mqttDeviceWarehouse -> {
                    if (mqttDeviceWarehouse != null && StringUtils.isNotEmpty(mqttDeviceWarehouse.getDevice_id())) {
                        return Mono.just(MqttDeviceInfoInCache.builder()
                                .domain(domain)
                                .subDomain(mqttDeviceWarehouse.getSub_domain())
                                .deviceType(mqttDeviceWarehouse.getDevice_type())
                                .status(mqttDeviceWarehouse.getStatus())
                                .deviceType(mqttDeviceWarehouse.getDevice_type())
                                .publicKey(mqttDeviceWarehouse.getPublic_key())
                                .build()).zipWhen(newInfo -> {
                            mqttDeviceInfoInCaches.saveToCache(wifiId, newInfo);
                            return Mono.just(newInfo);
                        }, (x, y) -> y);
                    }
                    return Mono.just(MqttDeviceInfoInCache.builder().build());
                }, (x, y) -> y);
            }
        }, (x, y) -> y);
    }

    private boolean validate(MqttDeviceInfoInCache inCache){
        return StringUtils.isNotEmpty(inCache.getWifiId()) && StringUtils.isNotEmpty(inCache.getSubDomain());
    }

    private Mono<ProductInfoInCache> fetchProductInfoInCaches(String domain, String subDomain) {
        return Mono.just(Optional.ofNullable(productInfoInCaches.getFromCache(String.format("%s-%s", domain, subDomain)))
                .orElse(ProductInfoInCache.builder().build())).zipWhen(inCache -> {
            if (0 != inCache.getMsgTransType()) {
                return Mono.just(inCache);
            } else {
                return Mono.just(productDao.findProductInfo(domain, subDomain)).or(Mono.just(new ProductInfo())).zipWhen(productInfo -> {
                    if (productInfo != null && productInfo.getMsgTransType() != 0) {
                        return Mono.just(ProductInfoInCache.builder()
                                .msgTransType(productInfo.getMsgTransType())
                                .oneMachineOneSecret(productInfo.getOneMachineOneSecret())
                                .communication(productInfo.getCommunication())
                                .build()).zipWhen(newInfo -> {
                            productInfoInCaches.saveToCache(String.format("%s-%s", domain, subDomain), newInfo);
                            return Mono.just(newInfo);
                        }, (x, y) -> y);
                    }
                    return Mono.just(ProductInfoInCache.builder().build());
                }, (x, y) -> y);
            }
        }, (x, y) -> y);
    }

    private boolean updateStatusInCache(String wifiId) {
        MqttDeviceInfoInCache deviceInfoInCacheOld = mqttDeviceInfoInCaches.getFromCache(wifiId);
        if (null != deviceInfoInCacheOld) {
            deviceInfoInCacheOld.setStatus(Constants.OldDays.ACTIVE);
            mqttDeviceInfoInCaches.saveToCache(wifiId, deviceInfoInCacheOld);
        }
        return true;
    }

    private String zipSql(String domainId, List<MqttDeviceWarehouse> req) {
        return Constants.Warehouse.INSERT_PREFIX
                .replaceAll(Constants.Warehouse.DOMAIN_ID_TO_REPLACE, domainId)
                + req.stream().map(info -> String.format("('%s', '%s', %d, '%s', %d, %d, '%s', UTC_TIMESTAMP(), UTC_TIMESTAMP())",
                info.getSub_domain(),
                info.getDevice_id(),
                Constants.OldDays.MASTER,
                info.getPublic_key(),
                Constants.OldDays.ACTIVE,
                info.getCode(),
                info.getBatch_code()
        )).collect(Collectors.joining(",")) + Constants.Warehouse.INSERT_SUFFIX;
    }
}
